package com.example.hbase.annotation;

import com.example.hbase.exception.HBaseException;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.util.Bytes;

import java.lang.reflect.Field;

/**
 * POJO的字段和hbase表的列映射信息
 */
public class ColumnInfo {

    /** java实体类类型 */
    private Class<?> type;
    /** 实体类字段 */
    private Field field;
    /** hbase的列簇-字符串形式 */
    private String family;
    /** hbase的列簇-字节形式. */
    private byte[] familyBytes;
    /** hbase的列-字符串形式 */
    private String qualifier;
    /** hbase的列-字节形式 */
    private byte[] qualifierBytes;
    /** 是否为hbase的rowKey */
    private boolean isRowKey;

    public String getFamily() {
        return family;
    }

    public String getQualifier() {
        return qualifier;
    }

    public boolean isRowKey() {
        return isRowKey;
    }

    /**
     * 从实体类的字段解析列信息
     * @param type POJO's class type.
     * @param field POJO' field.
     * @return ColumnInfo.
     * */
    static ColumnInfo parse(Class<?> type, Field field) {
        String defaultFamily = null;

        HBaseRowKey hbaseRowkey = field.getAnnotation(HBaseRowKey.class);
        if(hbaseRowkey!=null){
            ColumnInfo columnInfo = new ColumnInfo();
            columnInfo.type = type;
            columnInfo.field = field;
            columnInfo.isRowKey = true;
            return columnInfo;
        }

        HBaseTable hbaseTable = type.getAnnotation(HBaseTable.class);
        if (hbaseTable != null) {
            defaultFamily = hbaseTable.defaultFamily();
        }

        HBaseColumn hbaseColumn = field.getAnnotation(HBaseColumn.class);
        if (hbaseColumn == null) {
            return null;
        }

        //列簇如果没有指定，就用默认的
        String family = StringUtils.isBlank(hbaseColumn.family()) ? defaultFamily : hbaseColumn.family();
        String qualifier = hbaseColumn.qualifier();

        if (StringUtils.isBlank(family)) {
            throw new HBaseException("family is null or empty. type="
                    + type + " field=" + field);
        }
        if (StringUtils.isBlank(qualifier)) {
            throw new HBaseException("qualifier is null or empty. type="
                    + type + " field=" + field);
        }

        ColumnInfo columnInfo = new ColumnInfo();
        columnInfo.type = type;
        columnInfo.field = field;
        columnInfo.family = family;
        columnInfo.familyBytes = Bytes.toBytes(family);
        columnInfo.qualifier = qualifier;
        columnInfo.qualifierBytes = Bytes.toBytes(qualifier);
        columnInfo.isRowKey = false;
        return columnInfo;
    }

    private ColumnInfo() {
    }

    public Field getField() {
        return field;
    }

    @Override
    public String toString() {
        return "type=" + type + " field=" + field + " family=" + family
                + " qualifier=" + qualifier;
    }
}